package com.rpcdemo.rpc.transport;

import com.example.serialization.SeriaUtil;
import com.rpcdemo.rpc.ResponseMappingHandler;
import com.rpcdemo.rpc.protocol.MyContent;
import com.rpcdemo.rpc.protocol.MyHeader;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.*;

import java.io.*;
import java.net.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @ClassName ClientFactory
 * @Description TODO
 * @Author: XMK
 * @Date: 2020/9/30 9:47
 * @Version: 1.0
 */
public class ClientFactory {
    private int poolSize = 5;

    private static class ClientFactoryHolder{
        public static ClientFactory factory = new ClientFactory();
    }

    private ConcurrentHashMap<InetSocketAddress, ClientPool> poolHashMap = new ConcurrentHashMap<>();

    private ClientFactory() {
    }

    public static ClientFactory getFactory() {
        return ClientFactoryHolder.factory;
    }

    public CompletableFuture<Object> transport(MyContent content) {

        String type = "http";
//        String type = "rpc";

        CompletableFuture<Object> cmf = new CompletableFuture<>();
        if ("rpc".equals(type)) {


            byte[] bodybytes = SeriaUtil.encode(content);
            MyHeader header = MyHeader.createHeader(bodybytes);
            byte[] headerbytes = SeriaUtil.encode(header);
//        System.out.println("header length:::" + headerbytes.length);

            ResponseMappingHandler.addCallback(header.getRequestID(), cmf);

            // 连接池
            ByteBuf buf = ByteBufAllocator.DEFAULT.directBuffer(headerbytes.length + bodybytes.length);
            buf.writeBytes(headerbytes);
            buf.writeBytes(bodybytes);
            try {
                ClientFactory.getFactory()
                        .getClient(new InetSocketAddress("localhost", 9090))
                        .writeAndFlush(buf)
                        .sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }


        } else if ("http".equals(type)) {
            // 使用HTTP协议作为载体
            // 1. 用URL 现成的工具（包含了http的编解码，发送，socket，连接）
//            urlTS(content, cmf);

            // on netty， + netty提供的http相关的编解码
            nettyHttpTS(content, cmf);

        }
        return cmf;

    }

    private void nettyHttpTS(MyContent content, CompletableFuture<Object> cmf) {

        NioEventLoopGroup group = new NioEventLoopGroup(1);
        Bootstrap bs = new Bootstrap();
        Bootstrap client = bs.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new HttpClientCodec())
                                .addLast(new HttpObjectAggregator(1024 * 512))
                                .addLast(new ChannelInboundHandlerAdapter() {
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        // 接收
                                        FullHttpResponse response = (FullHttpResponse) msg;
                                        System.out.println(response.toString());

                                        ByteBuf byteBuf = response.content();
                                        byte[] bytes = new byte[byteBuf.readableBytes()];
                                        byteBuf.readBytes(bytes);
                                        ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
                                        MyContent o = (MyContent)ois.readObject();

                                        cmf.complete(o.getRes());
                                    }
                                });
                    }
                });

        try {
            byte[] bytes = SeriaUtil.encode(content);
            DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_0,
                    HttpMethod.POST,
                    "/",
                    Unpooled.copiedBuffer(bytes));

            request.headers().add(HttpHeaderNames.CONTENT_LENGTH, bytes.length);
            ChannelFuture localhost = client.connect("localhost", 9090).sync().channel().writeAndFlush(request).sync();


        } catch (InterruptedException e) {

            e.printStackTrace();
        }

    }

    private void urlTS(MyContent content, CompletableFuture<Object> cmf) {

        try {
            URL url = new URL("http://localhost:9090/");

            HttpURLConnection connection = (HttpURLConnection)url.openConnection();

            connection.setRequestMethod("POST");
            connection.setDoOutput(true);
            connection.setDoInput(true);

            connection.connect();
            OutputStream os = connection.getOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(os);
            oos.writeObject(content);

            os.flush();
            os.close();

            if (connection.getResponseCode() == 200) {

                InputStream is = connection.getInputStream();
                ObjectInputStream ois = new ObjectInputStream(is);
                MyContent respContent = (MyContent)ois.readObject();
                Object res = respContent.getRes();

                cmf.complete(res);
            }

        } catch (MalformedURLException e) {

        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }

    }

    public NioSocketChannel getClient(InetSocketAddress address) {

        ClientPool clientPool = poolHashMap.get(address);
        if (clientPool == null) {
            poolHashMap.putIfAbsent(address, new ClientPool(poolSize));
            clientPool = poolHashMap.get(address);
        }

        NioSocketChannel client = clientPool.getClient(address);

        return client;

    }
}
